package com.at.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/29 9:24 PM
 */
object SparkStreaming08_Close {

  def main(args: Array[String]): Unit = {
    /**
     * 线程的关闭
     * val thread = new Thread()
     * thread.start()
     *
     * thread.stop(); // 强制关闭
     */
    val sc = new SparkConf().setMaster("local[*]").setAppName("operator")
    val ssc = new StreamingContext(sc, Seconds(3))
    ssc.checkpoint("cp")

    val lines = ssc.socketTextStream("localhost", 9999)

    val word2One = lines.map((_, 1))
    word2One.print()

    ssc.start()
    // 如果想要关闭采集器，需要创建新的线程
    // 而且需要在第三方程序中增加关闭状态
    new Thread(new Runnable {
      override def run(): Unit = {

        // 优雅关闭
        // 计算节点不再接收新的数据，而是将现有数据执行完毕，再关闭
        // Mysql: Table(stopSpark) => row => data
        // Redis: Data(K-V)
        // ZK:  /stopSpark
        // HDFS: /stopSpark
        while (true) {
          println("1111111111111111111111111111")
          Thread.sleep(5000)
          if (true) {
            // 获取SparkStreaming状态
            val state = ssc.getState()
            if (state == StreamingContextState.ACTIVE) {
              println("222222222222222222222222222")
              ssc.stop(true, true)
              System.exit(0)
            }
          }
        }
      }
    }).start()

    ssc.awaitTermination() // 阻塞main线程

  }

}
